package com.changdu.seatunnel.admin.util;

import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.List;

import com.changdu.seatunnel.admin.entity.DataInstance;
import com.changdu.seatunnel.admin.pojo.TableField;

import lombok.extern.slf4j.Slf4j;

@Slf4j
public class DatabaseUtil {
    
	private static Connection getConnection(DataInstance config, String database) throws SQLException {
        return DriverManager.getConnection(buildJdbcUrl(config, database), config.getUsername(), config.getPassword());
    }	
	
	/**
	 * 获取库下的表
	 * @param config
	 * @return
	 */
    public static List<String> getTables(DataInstance config, String database) {
        List<String> tables = new ArrayList<>();
        try (Connection conn = getConnection(config, database);
             Statement stmt = conn.createStatement()) {
            
            String sql = "SHOW TABLES";
            if ("MySQL".equalsIgnoreCase(config.getType())) {
                sql = "SHOW TABLES FROM " + database;
            } else if ("PostgreSQL".equalsIgnoreCase(config.getType())) {
                sql = "SELECT table_name FROM information_schema.tables WHERE table_schema = 'public'";
            }
            
            try (ResultSet rs = stmt.executeQuery(sql)) {
                while (rs.next()) {
                    tables.add(rs.getString(1));
                }
            }
        } catch (Exception e) {
            log.error("获取表列表失败", e);
        }
        
        return tables;
    }
    
    private static String buildJdbcUrl(DataInstance config, String database) {
        String url;
        switch (config.getType().toLowerCase()) {
        	case "starrocks":
            case "mysql":
                url = String.format("jdbc:mysql://%s:%d/%s?useUnicode=true&characterEncoding=utf8&useSSL=false&serverTimezone=Asia/Shanghai",
                        config.getHost(), config.getPort(), database);
                break;
            default:
                throw new IllegalArgumentException("Unsupported database type: " + config.getType());
        }
        return url;
    }

    /**
     * 获取表结构
     * @param config
     * @param table
     * @return
     */
	public static List<TableField> getTableStructure(DataInstance config, String database, String table) {
        List<TableField> columns = new ArrayList<>();
        try (Connection conn = getConnection(config, database); Statement stmt = conn.createStatement()) {
            DatabaseMetaData metaData = conn.getMetaData();
            ResultSet rs = metaData.getColumns(database, null, table, null);
            while (rs.next()) {
            	TableField field = new TableField();
            	field.setFieldName(rs.getString("COLUMN_NAME"));
            	field.setFieldType(rs.getString("TYPE_NAME"));
            	field.setComment(rs.getString("REMARKS"));
            	field.setDefaultValue(rs.getString("COLUMN_DEF"));
            	field.setNullable(rs.getBoolean("IS_NULLABLE"));
                columns.add(field);
            }
        } catch (SQLException e) {
            log.error("获取表结构失败", e);
        }
        return columns;
	}

	/**
	 * 获取所有数据库
	 * 
	 * @param config
	 * @return
	 */
	public static List<String> getDatabases(DataInstance config) {
		List<String> databases = new ArrayList<>();
		try (Connection conn = getConnection(config, ""); Statement stmt = conn.createStatement()) {
			DatabaseMetaData metaData = conn.getMetaData();
			ResultSet rs = metaData.getCatalogs();
			while (rs.next()) {
				databases.add(rs.getString(1));
			}
		} catch (SQLException e) {
			log.error("获取数据库列表失败", e);
		}
		return databases;
	}
	
	/**
	 * 从后往前遍历字符，找出最前面连续的数字的索引位置
	 * @param tableName
	 * @return
	 */
	public static int findRegexIndex(String tableName) {
		for (int i = tableName.length() - 1; i >= 0; i--) {
		    if (Character.isDigit(tableName.charAt(i))) {
		        continue;
		    }
		    return i + 1;
		}
		return tableName.length();
	}	
} 